  WITH
   scan         AS (-- 揽收 taking
                    SELECT billcode           AS waybill_no,
                           scansitecode_early AS network_code,
                           scantime_early     AS scan_time,
                           NULL               AS next_network_code,
                           NULL               AS shipment_no,
                           FALSE              AS is_center_arrival,
                           FALSE              AS is_deliver
                      FROM jms_dwd.dwd_barscan_taking_dt
                     WHERE dt BETWEEN date_add('{{ execution_date | cst_ds }}', -6) AND '{{ execution_date | cst_ds }}'
                     UNION ALL
                    -- 入仓 warehousing
                    SELECT billcode           AS waybill_no,
                           scansitecode_early AS network_code,
                           scantime_early     AS scan_time,
                           NULL               AS next_network_code,
                           NULL               AS shipment_no,
                           FALSE              AS is_center_arrival,
                           FALSE              AS is_deliver
                      FROM jms_dwd.dwd_barscan_warehousing_dt
                     WHERE dt BETWEEN date_add('{{ execution_date | cst_ds }}', -6) AND '{{ execution_date | cst_ds }}'
                       AND dimtype = 'billcode'
                     UNION ALL
                    -- 装车 loading
                    SELECT billcode                          AS waybill_no,
                           scansitecode_early                AS network_code,
                           scantime_early                    AS scan_time,
                           nullif(nextstationcode_early, '') AS next_network_code,
                           nullif(transfercode_early, '')    AS shipment_no,
                           FALSE                             AS is_center_arrival,
                           FALSE                             AS is_deliver
                      FROM jms_dwd.dwd_barscan_loading_dt
                     WHERE dt BETWEEN date_add('{{ execution_date | cst_ds }}', -6) AND '{{ execution_date | cst_ds }}'
                     UNION ALL
                    -- 发件 send
                    SELECT billcode                          AS waybill_no,
                           scansitecode_early                AS network_code,
                           scantime_early                    AS scan_time,
                           nullif(nextstationcode_early, '') AS next_network_code,
                           nullif(transfercode_early, '')    AS shipment_no,
                           FALSE                             AS is_center_arrival,
                           FALSE                             AS is_deliver
                      FROM jms_dwd.dwd_barscan_sitesend_dt
                     WHERE dt BETWEEN date_add('{{ execution_date | cst_ds }}', -6) AND '{{ execution_date | cst_ds }}'
                     UNION ALL
                    SELECT billcode                          AS waybill_no,
                           scansitecode_early                AS network_code,
                           scantime_early                    AS scan_time,
                           nullif(nextstationcode_early, '') AS next_network_code,
                           nullif(transfercode_early, '')    AS shipment_no,
                           FALSE                             AS is_center_arrival,
                           FALSE                             AS is_deliver
                      FROM jms_dwd.dwd_barscan_centersend_dt
                     WHERE dt BETWEEN date_add('{{ execution_date | cst_ds }}', -6) AND '{{ execution_date | cst_ds }}'
                     UNION ALL
                    -- 到件 arrival
                    SELECT billcode           AS waybill_no,
                           scansitecode_early AS network_code,
                           scantime_early     AS scan_time,
                           NULL               AS next_network_code,
                           NULL               AS shipment_no,
                           FALSE              AS is_center_arrival,
                           FALSE              AS is_deliver
                      FROM jms_dwd.dwd_barscan_sitearrival_dt
                     WHERE dt BETWEEN date_add('{{ execution_date | cst_ds }}', -6) AND '{{ execution_date | cst_ds }}'
                     UNION ALL
                    SELECT billcode           AS waybill_no,
                           scansitecode_early AS network_code,
                           scantime_early     AS scan_time,
                           NULL               AS next_network_code,
                           NULL               AS shipment_no,
                           TRUE               AS is_center_arrival,
                           FALSE              AS is_deliver
                      FROM jms_dwd.dwd_barscan_centerarrival_dt
                     WHERE dt BETWEEN date_add('{{ execution_date | cst_ds }}', -6) AND '{{ execution_date | cst_ds }}'
                     UNION ALL
                    -- 派件 deliver
                    SELECT billcode           AS waybill_no,
                           scansitecode_early AS network_code,
                           scantime_early     AS scan_time,
                           NULL               AS next_network_code,
                           NULL               AS shipment_no,
                           FALSE              AS is_center_arrival,
                           TRUE               AS is_deliver
                      FROM jms_dwd.dwd_barscan_deliver_dt
                     WHERE dt BETWEEN date_add('{{ execution_date | cst_ds }}', -6) AND '{{ execution_date | cst_ds }}'
                       AND dimtype = 'billcode'),
   -- 各网点最早 揽收/入仓/装车/发件/到件/出仓 时间
   earlist_scan AS (SELECT scan.waybill_no                           AS waybill_no,
                           scan.network_code                         AS network_code,
                           min(scan.scan_time)                       AS earliest_time,
                           max(scan.is_deliver)                      AS has_deliver,
                           max(scan.is_center_arrival)               AS is_center_arrival,
                           first_value(scan.next_network_code, TRUE) AS next_network_code,
                           first_value(scan.shipment_no, TRUE)       AS shipment_no
                      FROM scan
                      LEFT SEMI JOIN jms_ods.yl_lmdm_sys_network AS network
                        ON network.dt = '{{ execution_date | cst_ds }}'
                       AND network.code = scan.network_code
                       AND (network.type_id != 335 OR (network.type_id = 335 AND scan.is_center_arrival))
                     WHERE scan.network_code != '888888'
                       AND scan.network_code IS NOT NULL
                       AND scan.network_code != ''
                       AND length(scan.waybill_no) <= 15
                       AND scan.waybill_no NOT LIKE 'B%'
                     GROUP BY scan.waybill_no, scan.network_code),
       -- 运输车辆信息
       shipment AS (SELECT shipment.shipment_no                           AS shipment_no,             -- 转运任务单号
                           shipment.start_code                            AS start_network_code,      -- 发车网点编码
                           shipment.end_code                              AS end_network_code,        -- 到车网点编码
                           first_value(shipment.end_id, TRUE)             AS end_network_id,          -- 到车网点ID
                           first_value(shipment.end_name, TRUE)           AS end_network_name,        -- 到车网点名称
                           first_value(end_network.type_id, TRUE)         AS end_network_type_id,     -- 到车网点类型
                           first_value(end_network.is_entrepot = 1, TRUE) AS end_network_is_entrepot, -- 到车网点是否为集散点
                           first_value(shipment.line_id, TRUE)            AS line_id,                 -- 车线ID
                           first_value(shipment.line_code, TRUE)          AS line_code,               -- 车线编码
                           first_value(shipment.line_name, TRUE)          AS line_name,               -- 车线名称
                           max(timestamp(shipment.load_time))             AS load_time,               -- 装车时间
                           max(timestamp(shipment.lock_time))             AS lock_time,               -- 锁车时间
                           min(timestamp(shipment.actual_arrival_time))   AS actual_arrival_time,     -- 实际到达时间
                           min(timestamp(shipment.planned_arrival_time))  AS planned_arrival_time     -- 计划到达时间
                      FROM jms_ods.yl_tms_shipment AS shipment
                      LEFT JOIN jms_ods.yl_lmdm_sys_network AS end_network ON end_network.dt = '{{ execution_date | cst_ds }}' AND end_network.code = shipment.end_code
                     WHERE shipment.dt = '{{ execution_date | cst_ds }}'
                       AND date(timestamp(actual_departure_time)) >= date_add('{{ execution_date | cst_ds }}', -2)
                     GROUP BY shipment_no, start_network_code, end_network_code),
       -- 运单所经网点按时间排序
       transfer AS (SELECT waybill_no                                                                  AS waybill_no,
                           earliest_time                                                               AS earliest_time,
                           lead(earliest_time) OVER ordered_window                                     AS lead_earliest_time,
                           network_code                                                                AS network_code,
                           lead(network_code) OVER ordered_window                                      AS lead_network_code,
                           next_network_code                                                           AS next_network_code,
                           shipment_no                                                                 AS shipment_no,
                           row_number() OVER ordered_window                                            AS step,
                           max(network.type_id = 335) OVER ordered_window                              AS is_after_center,
                           max(has_deliver) OVER full_window                                           AS has_deliver,
                           max(is_center_arrival) OVER full_window                                     AS has_center_arrival,
                           network.id                                                                  AS network_id,
                           network.type_id                                                             AS network_type_id,
                           network.type_id = 336 AND network.is_entrepot = 1                           AS network_is_entrepot,
                           network.name                                                                AS network_name,
                           lead(network.id) OVER ordered_window                                        AS lead_network_id,
                           lead(network.type_id) OVER ordered_window                                   AS lead_network_type_id,
                           lead(network.type_id = 336 AND network.is_entrepot = 1) OVER ordered_window AS lead_network_is_entrepot,
                           lead(network.name) OVER ordered_window                                      AS lead_network_name
                      FROM earlist_scan
                      LEFT JOIN jms_ods.yl_lmdm_sys_network AS network
                        ON network.dt = '{{ execution_date | cst_ds }}'
                       AND network.code = network_code
                    WINDOW ordered_window AS (PARTITION BY waybill_no ORDER BY earliest_time),
                           full_window AS (PARTITION BY waybill_no)),
       collect_transfer AS (SELECT transfer.waybill_no                                                                         AS waybill_no,                    -- 运单号
                                   transfer.earliest_time                                                                      AS send_time,                     -- 发件扫描时间
                                   nvl(shipment.actual_arrival_time, transfer.lead_earliest_time)                              AS arrival_time,                  -- 到件扫描时间
                                   transfer.network_id                                                                         AS from_network_id,               -- 发件网点ID
                                   transfer.network_code                                                                       AS from_network_code,             -- 发件网点编码
                                   transfer.network_name                                                                       AS from_network_name,             -- 发件网点名称
                                   transfer.network_type_id                                                                    AS from_network_type_id,          -- 发件网点类型
                                   transfer.network_is_entrepot                                                                AS from_network_is_entrepot,      -- 发件网点是否为集散点
                                   nvl(transfer.lead_network_id, shipment.end_network_id)                                      AS to_network_id,                 -- 到件网点ID
                                   coalesce(transfer.lead_network_code, transfer.next_network_code, shipment.end_network_code) AS to_network_code,               -- 到件网点编码
                                   nvl(transfer.lead_network_name, shipment.end_network_name)                                  AS to_network_name,               -- 到件网点名称
                                   nvl(transfer.lead_network_type_id, shipment.end_network_type_id)                            AS to_network_type_id,            -- 到件网点类型
                                   nvl(transfer.lead_network_is_entrepot, shipment.end_network_is_entrepot)                    AS to_network_is_entrepot,        -- 到件网点是否为集散点
                                   shipment.shipment_no                                                                        AS shipment_no,                   -- 转运任务单号
                                   shipment.line_id                                                                            AS line_id,                       -- 车线ID
                                   shipment.line_code                                                                          AS line_code,                     -- 车线编码
                                   shipment.line_name                                                                          AS line_name,                     -- 车线名称
                                   shipment.load_time                                                                          AS shipment_load_time,            -- 装车时间
                                   shipment.lock_time                                                                          AS shipment_lock_time,            -- 锁车时间
                                   shipment.actual_arrival_time                                                                AS shipment_arrival_time,         -- 车辆实际到达时间
                                   shipment.planned_arrival_time                                                               AS shipment_planned_arrival_time, -- 车辆计划到达时间
                                   transfer.step                                                                               AS step,                          -- 第几次集货转运
                                   sum(1) OVER (PARTITION BY transfer.waybill_no)                                              AS full_step,                     -- 集货流程一共经历了几次转运
                                   date(transfer.earliest_time)                                                                AS dt
                              FROM transfer
                              LEFT JOIN shipment
                                ON transfer.lead_network_code = transfer.next_network_code
                               AND transfer.shipment_no IS NOT NULL
                               AND shipment.start_network_code = transfer.network_code
                               AND shipment.end_network_code = transfer.lead_network_code
                               AND shipment.shipment_no = transfer.shipment_no
                             WHERE transfer.is_after_center = FALSE             -- 排除经过转运中心之后的所有记录
                               AND NOT (has_deliver AND NOT has_center_arrival) -- 排除自派件：有出仓记录且无中心到件记录
                               AND NOT (transfer.network_is_entrepot AND transfer.lead_network_type_id = 336 AND NOT transfer.lead_network_is_entrepot) -- 排除集散到网点记录
       )
INSERT OVERWRITE TABLE dm.dm_collect_trace_dt PARTITION (dt)
SELECT *
  FROM collect_transfer
 -- 如果运行当天任务，则动态生成最新的2个分区，否则仅生成1个分区
 WHERE dt BETWEEN date_add('{{ execution_date | cst_ds }}', -1)
              AND date_add('{{ execution_date | cst_ds }}', if(datediff(current_timestamp(), '{{ execution_date | cst_ds }}') > 1, -1, 0))
DISTRIBUTE BY pmod(hash(waybill_no), 6);
